/*
 *  Copyright 2020-2025 the original author or authors.
 *  You cannot use this file unless authorized by the author.
 */

package org.ipig.computing.spark.streaming;

import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.streaming.api.java.JavaInputDStream;
import org.ipig.commons.helper.DateTimeHelper;
import org.ipig.computing.constant.context.SparkContext;

/**
 * RDD 处理
 *
 * @author <a href="mailto:comchnts@163.com">chinats</a>
 * @since 1.0
 */
public interface SparkStreamRddService<T> extends SparkStreamService<T> {
    @Override
    default void doStream(JavaInputDStream<T> stream) {
/**
 获取偏移量
 stream.foreachRDD(rdd -> {
 OffsetRange[] offsetRanges = ((HasOffsetRanges) rdd.rdd()).offsetRanges();
 rdd.foreachPartition(consumerRecords -> {
 OffsetRange o = offsetRanges[TaskContext.get().partitionId()];
 System.out.println(o.topic() + " " + o.partition() + " " + o.fromOffset() + " " + o.untilOffset());
 });
 });
 */
        stream.foreachRDD(rdd -> {
/*            OffsetRange[] offsetRanges = ((HasOffsetRanges) rdd.rdd()).offsetRanges();
            for (OffsetRange offsetRange:offsetRanges){
                System.out.println(offsetRange.topicPartition().toString()+" ; "+offsetRange.toString());
            }
            rdd.foreachPartition(consumerRecords -> {
                OffsetRange offsetRange = offsetRanges[TaskContext.get().partitionId()];
                System.out.println("主题"+offsetRange.topic() + " 分区" + offsetRange.partition() + " 开始位置" + offsetRange.fromOffset() + " 当前位置" + offsetRange.untilOffset());
            });*/
            long count=rdd.count();//在同一时间间隔里， 只返回一个RDD的“微批次”
            System.out.printf("thread=%s，time=%s,allCount=%s\n",this.getSession().conf().get(SparkContext.Application.APP_NAME.key), DateTimeHelper.getDateTime(),count);
//            if (count >0) {
            try {
                //初始化每个RDD
                initEachRDD(rdd);
            } catch (Exception e) {
                e.printStackTrace();
                //RDD 初始失败就直接结束程序
                System.exit(0);
            }
            try {
                //处理每个RDD
                doEachRDD(rdd);
            } catch (Exception e) {
                e.printStackTrace();
            }
            try {
                //每个RDD处理完成后的回调
                postEachRDD(rdd);
            } catch (Exception e) {
                e.printStackTrace();
            }
//            }
        });
    }


    /**
     * 初始化每个RDD
     */
    public default void initEachRDD(JavaRDD<T> rdd) {

    }

    /**
     * 处理每个RDD
     *
     * @param rdd JavaRDD<T>
     */
    public void doEachRDD(final JavaRDD<T> rdd);

    /**
     * 每个RDD处理完成后的回调
     * <li>此方法在Driver端执行，可以进行累加器值的获取等</li>
     */
    public void postEachRDD(final JavaRDD<T> rdd);
}
